package com.atguigu.api4

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, Table}
import org.apache.flink.table.descriptors.{Csv, Kafka, Schema}

/**
 * @description: 流式输出
 * @time: 2020/7/22 17:22
 * @author: baojinlong
 **/
object kafkaTableTest {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度和table环境配置
    environment.setParallelism(1)
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useOldPlanner()
      .inStreamingMode()
      .build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(environment, settings)

    // 从kafka读取数据
    // 连接到Kafka
    tableEnv.connect(
      new Kafka()
        .version("0.11")
        .topic("sensor")
        .property("bootstrap.servers", "localhost:9092")
        .property("zookeeper.connect", "localhost:2182")
    )
      .withFormat(new Csv)
      .withSchema(
        new Schema()
          .field("id", DataTypes.STRING)
          .field("timestamp", DataTypes.BIGINT)
          .field("temperature", DataTypes.DOUBLE)
      )
      .createTemporaryTable("kafkaInputTable")
    // 做转换操作
    val sensorTable: Table = tableEnv.from("kafkaInputTable")
    // 做转换操作
    // 对table进行操作得到结果表
    val resultTable: Table = sensorTable
      .select('id, 'temperature)
      .filter('id === "sensor_01")
    // 聚合操作得到表数据
    val aggResultTable: Table = sensorTable
      .groupBy('id)
      .select('id, 'id.count as 'cnt)

    // 定义一个连接到kafka连接表
    tableEnv.connect(
      new Kafka()
        .version("0.11")
        .topic("sensorTest")
        .property("bootstrap.servers", "localhost:9092")
        .property("zookeeper.connect", "localhost:2182")
    )
      .withFormat(new Csv)
      .withSchema(
        new Schema()
          .field("id", DataTypes.STRING)
          // 字段名可以不一致
          .field("temp", DataTypes.DOUBLE)
      )
      .createTemporaryTable("kafkaOutputTable")

    // 将结果表输出,同样只能输出是Append才可以,聚合的不能输出
    resultTable.insertInto("kafkaOutputTable")

    // 执行
    environment.execute("OutputTableTest job")
  }

}
